package com.yeyouluo.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;

import java.io.IOException;
import java.util.Collection;
import java.util.List;

public class MyDeleteProcessor extends BaseRegionObserver {

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        try (HTableInterface relation = e.getEnvironment().getTable(TableName.valueOf(CreateHBaseTable.TABLE_NAME))) {
            Collection<List<Cell>> values = delete.getFamilyCellMap().values();
            for (List<Cell> value : values) {
                for (Cell cell : value) {
                    byte[] rawRow = CellUtil.cloneRow(cell);
                    byte[] rawCF = CellUtil.cloneFamily(cell);
                    byte[] rawCQ = CellUtil.cloneQualifier(cell);

                    // 防止递归触发造成资源耗尽
                    boolean exists = relation.exists(new Get(rawCQ).addColumn(rawCF, rawRow));
                    if (exists) {
                        Delete delete2 = new Delete(rawCQ).addColumn(rawCF, rawRow);
                        relation.delete(delete2);
                    }
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
